package org.chnxi.study.messagequeue.kafka.comsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class CommitSyncConsumer {

    public Properties kafkaConsumerProps(){
        Properties props = new Properties();
        // 设置kafka服务地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
        // 设置session超时时长
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
        // 心跳检测间隔，受限于消息处理时长，要大于消息处理时长
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 60000);
        // 设置消费者组名称
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroup");
        // 设置key和value的反序列化处理器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关闭自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 设置初始化offset的方式
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 设置每次最大拉取消息数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        return props;
    }

    public void consumer(){
        Properties props = kafkaConsumerProps();
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);

        kafkaConsumer.subscribe(Collections.singletonList("topicName"));

        try{
            while (true){
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100).toMillis());

                // TODO 增加records的处理程序

                // 设置同步提交
                kafkaConsumer.commitSync();
            }
        } catch (Exception e){
            // 消费失败。。。
        }finally {
            kafkaConsumer.close();
        }

    }

}
